"""
Case Type   : 数据库系统
Case Name   : 1主1备1级联,备节点事务提交方式为off
Create At   : 2021/02/24
Owner       : songjing20
Description :
    1.初始化环境
    2.将备节点1切换为级联备
    3.设置事务同步方式为off,synchronous_standby_names为空
    4.配置所有节点备份延迟为1min
    5.查询集群同步方式
    6.创建表并插入数据
    7.等待10s查询备机
    8.等待1min查询备机
    9.插入数据
    10.等待10s查询备机
    11.等待1min查询备机
    12.删除数据
    13.立即修改延迟参数修改为0
    14.备机查询,等待1min再查询
    15.清理环境
Expect      :
    1.成功
    2.切换成功
    3.设置成功
    4.设置成功
    5.级联备为异步,备节点为异步
    6.创建并插入成功
    7.查询不到
    8.数据同步
    9.数据插入成功
    10.备节点未同步
    11.备节点同步
    12.删除成功
    13.配置成功
    14.未同步,1min后同步
    15.成功
History     :
    modified by haomeng at 2025.1.17:优化用例增加等待时间以解决环境性能不足问题导致查询失败
"""
import unittest
import time
import os
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant
from testcase.utils.CommonSH import CommonSH

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == Primary_SH.get_node_num(),
                 '需主备环境,若为单机环境则不执行')
class RecoveryDelay(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("----this is setup----")
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.db_primary_user_node = Node('PrimaryDbUser')
        self.comshpri = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.com = Common()
        self.tb_name = 'tb_case0010'
        self.rootnodelist = ['Standby1Root', 'Standby2Root']

    def test_recovery_delay(self):
        text = '----step1:初始化环境 expect:成功----'
        self.log.info(text)
        self.log.info('----1.1获取节点数目----')
        self.nodelist = ['Standby1DbUser', 'Standby2DbUser']
        result = self.comshpri.get_db_cluster_status('detail')
        self.log.info(result)
        self.node_num = result.count('Standby Normal') + 1
        self.comshsta = []
        self.log.info(self.node_num)
        self.assertEqual(self.node_num, 3, '执行失败' + text)
        for i in range(int(self.node_num) - 1):
            self.comshsta.append(CommonSH(self.nodelist[i]))
        self.log.info('----1.2安装ntp服务----')
        install_cmd = "yum install -y ntp && " \
                      "systemctl enable ntpd && systemctl start ntpd"
        for i in range(int(self.node_num) - 1):
            db_standby_node = Node(node=self.rootnodelist[i])
            result = db_standby_node.sh(install_cmd).result()
            self.log.info(install_cmd)
            self.log.info(result)
        self.log.info('----1.3同步集群时间----')
        current = self.db_primary_user_node.sh(
            "date \"+%m/%d/%Y %H:%M:%S\"").result()
        self.log.info(current)
        datecmd = f'date -s "{current}"'
        for i in range(int(self.node_num) - 1):
            db_standby_node = Node(node=self.rootnodelist[i])
            result = db_standby_node.sh(datecmd).result()
            self.log.info(datecmd)
            self.log.info(result)

        text = '----step2:将备节点1切换为级联备 expect:切换成功----'
        self.log.info(text)
        self.log.info('----2.1将备节点1切换为级联备,初始化环境----')
        result = self.comshpri.get_db_cluster_status('detail')
        self.log.info(result)
        result = self.comshsta[0].build_standby('-M cascade_standby')
        self.assertIn(self.constant.REBUILD_SUCCESS_MSG, result)
        result = self.comshsta[0].exec_refresh_conf()
        self.assertTrue(result, '执行失败' + text)
        self.log.info('----2.2查询数据库状态----')
        result = self.comshpri.get_db_cluster_status('detail')
        self.log.info(result)
        if 'Cascade Normal' not in result and 'Stanby Normal' not in \
                result:
            result = self.comshpri.restart_db_cluster()
            self.log.info(result)
            self.assertTrue(result, '执行失败' + text)

        text = '----step3:设置synchronous_commit=off,synchronous_standby' \
               '_names为空 expect:设置成功----'
        self.log.info(text)
        self.result = self.com.show_param('synchronous_standby_names')
        if self.result != '':
            re = self.comshpri.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                "synchronous_standby_names=''")
            self.log.info(re)
            self.assertTrue(re, '执行失败' + text)
        self.result1 = self.com.show_param('synchronous_commit')
        if self.result1 != 'off':
            re = self.comshpri.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                'synchronous_commit=off')
            self.assertTrue(re, '执行失败' + text)

        text = '----step4:设置recovery_min_apply_delay=1min ' \
               'expect:设置成功----'
        self.log.info(text)
        self.result2 = self.com.show_param('recovery_min_apply_delay')
        if self.result2 != '1min':
            re = self.comshpri.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                'recovery_min_apply_delay=1min')
            self.assertTrue(re, '执行失败' + text)
        self.log.info('--------查询参数--------')
        result = self.comshpri.execut_db_sql('show synchronous_commit;')
        self.log.info(result)
        self.assertIn('off', result, '执行失败' + text)
        result = self.comshpri.execut_db_sql(
            'show recovery_min_apply_delay;')
        self.log.info(result)
        self.assertIn('1min', result, '执行失败' + text)

        text = '----step5:查询集群同步方式 expect:级联备为异步,' \
               '备节点为异步----'
        self.log.info(text)
        result = self.comshpri.get_db_cluster_status('detail')
        self.log.info(result)
        sql = "select * from pg_stat_replication;"
        result = self.comshpri.execut_db_sql(sql)
        self.log.info(result)
        self.assertIn('Async', result, '执行失败' + text)
        result = self.comshsta[1].execut_db_sql(sql)
        self.log.info(result)
        self.assertIn('Async', result, '执行失败' + text)
        self.log.info('----等待主备一致----')
        result = self.comshsta[1].check_data_consistency()
        self.assertTrue(result)
        for i in range(90):
            result = self.comshpri.check_cascade_standby_consistency()
            if result:
                break
            time.sleep(20)
        self.assertTrue(result)

        text = '----step6:创建表,并插入数据 expect:创建并插入成功----'
        self.log.info(text)
        sql = f"drop table if exists {self.tb_name};" \
            f"create table {self.tb_name}(i int, s char(10));" \
            f"insert into {self.tb_name} values(5,'test');"
        result = self.comshpri.execut_db_sql(sql)
        self.log.info(result)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, result,
                      '执行失败' + text)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败'
                      + text)

        text = '----step7:等待10秒,备机查询 expect:查询不到----'
        self.log.info(text)
        time.sleep(10)
        sql = f"select * from {self.tb_name};"
        for i in range(int(self.node_num) - 1):
            result = self.comshsta[i].execut_db_sql(sql)
            self.log.info(result)
            self.assertIn(self.constant.NOT_EXIST, result, '执行失败' +
                          text)

        text = '----step8:等待60s,查询备机 expect:数据同步----'
        self.log.info(text)
        time.sleep(60)
        sql = f"select * from {self.tb_name};"
        for exec_time in range(5):
            flag = 0
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                if '5 | test' in result:
                    flag += 1
            if flag > 1:
                break
            time.sleep(10)
            self.assertEqual(2, flag, '执行失败' + text)

        text = '----step9:插入数据 expect:数据插入成功----'
        self.log.info(text)
        sql = f"insert into {self.tb_name} values(3, 'test');"
        result = self.comshpri.execut_db_sql(sql)
        self.log.info(result)
        self.assertTrue(self.constant.INSERT_SUCCESS_MSG, result)

        text = '----step10:等待10s,查询备机 expect:备节点未同步----'
        self.log.info(text)
        time.sleep(10)
        sql = f"select * from {self.tb_name};" \
            f"select count(*) from {self.tb_name};"
        for i in range(int(self.node_num) - 1):
            result = self.comshsta[i].execut_db_sql(sql)
            self.log.info(result)
            self.assertNotIn('3 | test', result)
            self.assertIn('1', result)
            self.assertIn('5 | test', result)

        text = '----step11:等待60s查询备机 expect:备节点同步----'
        self.log.info(text)
        time.sleep(60)
        sql = f"select * from {self.tb_name};" \
            f"select count(*) from {self.tb_name};"
        for exec_time in range(5):
            flag = 0
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                if ('3 | test' in result) and ('2' in result) \
                        and ('5 | test' in result):
                    flag += 1
            if flag > 1:
                break
            time.sleep(10)
            self.assertEqual(2, flag, '执行失败' + text)

        text = '----step12:删除数据 expect:成功----'
        self.log.info(text)
        sql = f"delete from {self.tb_name};"
        result = self.comshpri.execut_db_sql(sql)
        self.log.info(result)
        self.assertIn('DELETE', result, '执行失败' + text)

        text = '----step13:设置recovery_min_apply_delay=0 ' \
               'expect:配置成功----'
        self.log.info(text)
        result = self.comshpri.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            'recovery_min_apply_delay=0')
        self.assertTrue(result, '执行失败' + text)

        text = '----step14:备机查询  expect:数据删除被同步----'
        self.log.info(text)
        time.sleep(5)
        sql = f"select * from {self.tb_name};"
        for i in range(int(self.node_num) - 1):
            result = self.comshsta[i].execut_db_sql(sql)
            self.log.info(result)
            self.assertIn('(0 rows)', result, '执行失败' + text)
        time.sleep(60)
        for exec_time in range(5):
            flag = 0
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                if '(0 rows)' in result:
                    flag += 1
            if flag > 1:
                break
            time.sleep(10)
            self.assertEqual(2, flag, '执行失败' + text)

    def tearDown(self):
        text = '----step15:清理环境 expect:成功----'
        self.log.info(text)
        self.log.info('-----删除表----')
        sql = f"drop table if exists {self.tb_name};"
        result1 = self.comshpri.execut_db_sql(sql)
        self.log.info(result1)

        self.log.info('-----还原集群----')
        if self.node_num > 2:
            self.comshsta[0].build_standby('-M standby')
            result2 = self.comshsta[0].exec_refresh_conf()
            self.log.info(result2)
        result3 = self.comshpri.get_db_cluster_status('detail')
        self.log.info(result3)

        self.log.info('----------还原参数-------')
        cmd = f"alter system set synchronous_standby_names to '{self.result}'"
        re1 = self.comshpri.execut_db_sql(cmd)
        self.log.info(re1)
        re2 = self.comshpri.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'synchronous_commit={self.result1}')
        self.log.info(re2)
        re3 = self.comshpri.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'recovery_min_apply_delay={self.result2}')
        self.log.info(re3)

        self.assertIn('ALTER', re1, '执行失败' + text)
        self.assertTrue(re2, '执行失败' + text)
        self.assertTrue(re3, '执行失败' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1, '执行失败' +
                      text)
        self.assertEqual(result3.count('Standby Normal'), 2, '执行失败' + text)
        self.log.info(f'----{os.path.basename(__file__)} end----')
